package com.tju.shortlink.project.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.rabbitmq.client.Channel;
import com.tju.shortlink.project.common.constant.MQConstans;
import com.tju.shortlink.project.common.convention.exception.ServiceException;
import com.tju.shortlink.project.dao.entity.*;
import com.tju.shortlink.project.dao.mapper.*;
import com.tju.shortlink.project.dto.biz.BinLogEntity;
import com.tju.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.tju.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.amqp.core.Message;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;

import static com.tju.shortlink.project.common.constant.RedisKeyConstant.*;
import static com.tju.shortlink.project.common.constant.ShortLinkConstant.AMAP_REMOTE_URL;

@Component
@Slf4j
@RequiredArgsConstructor
public class RabbitMqShortLinkStatsSaveConsumer {

    private final LinkMapper shortLinkMapper;
    private final LinkGotoMapper shortLinkGotoMapper;
    private final RedissonClient redissonClient;
    private final LinkAccessStatsMapper linkAccessStatsMapper;
    private final LinkLocaleStatsMapper linkLocaleStatsMapper;
    private final LinkOsStatsMapper linkOsStatsMapper;
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;
    private final LinkAccessLogsMapper linkAccessLogsMapper;
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;
    private final LinkStatsTodayMapper linkStatsTodayMapper;
    private final StringRedisTemplate redisTemplate;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    @Value("${short-link.stats.locale.amap-key}")
    private String statsLocaleAmapKey;

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "canal.queue", durable = "true"),
            exchange = @Exchange(value = MQConstans.CANAL_EXCHANGE, type = ExchangeTypes.DIRECT),
            key = MQConstans.CANAL_TOPIC))
    public void deleteRedisCache(Channel channel, Message message) throws IOException {
        try {
            // 获取消息内容
            String content = new String(message.getBody(), StandardCharsets.UTF_8);
            // 反序列化
            BinLogEntity binLog = JSON.parseObject(content, BinLogEntity.class);

            // 根据操作类型和表名判断是否需要处理
            if (isRelevantTable(binLog.getTable()) && ("INSERT".equals(binLog.getType()) || "UPDATE".equals(binLog.getType()))) {
                List<LinkDo> data = binLog.getData(LinkDo.class);
                for(LinkDo link: data){
                    String fullShortUrl = link.getFullShortUrl();
                    if (fullShortUrl != null) {
                        // 删除Redis缓存 保证缓存一致性
                        redisTemplate.delete(String.format(GOTO_SHORT_LINK_KEY, fullShortUrl));
                        redisTemplate.delete(String.format(GOTO_IS_NULL_SHORT_LINK_KEY, fullShortUrl));
                    }
                }
            }

            // 确认消息已处理
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            // 处理异常，重新投递消息
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            throw new RuntimeException("消息处理失败", e);
        }
    }

    private boolean isRelevantTable(String tableName) {
        for (int i = 0; i <= 15; i++) {
            if (tableName.equals("t_link_" + i)) {
                return true;
            }
        }
        return false;
    }

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "stats.save", durable = "true"),
            exchange = @Exchange(value = MQConstans.PROJECT_EXCHANGE, type = ExchangeTypes.TOPIC),
            key = MQConstans.STATS_SAVE_TOPIC))
    public void onMessage(Channel channel, Message message) throws IOException {
        MessageProperties properties = message.getMessageProperties();
        String id = properties.getMessageId();

        String body = new String(message.getBody());
        Map<String, String> producerMap = JSON.parseObject(body, Map.class);

        log.info("监听到消息：{}", producerMap);
        // 1. 幂等性判断防止重复消费
        if (messageQueueIdempotentHandler.isMessageBeingConsumed(id.toString())) {
            // 1.1 判断当前的这个消息流程是否执行完成
            if (messageQueueIdempotentHandler.isAccomplish(id.toString())) {
                return;
            }
            // 1.2 消息正在处理中， 后续再进行投递
            throw new ServiceException("消息未完成流程，需要消息队列重试");
        }
        // 2. 处理消息
        try {
            ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
            // 2.2 统计信息
            actualSaveShortLinkStats(statsRecord);
            // 2.3 确认消息
            channel.basicAck(properties.getDeliveryTag(), false);
        } catch (Throwable ex) {
            // 宕机了 需要重新消费删除幂等组件
            channel.basicNack(properties.getDeliveryTag(), false, true);
            messageQueueIdempotentHandler.delMessageProcessed(id.toString());
            log.error("记录短链接监控消费异常", ex);
            throw ex;
        }
        // 3. 设置为完成状态
        messageQueueIdempotentHandler.setAccomplish(id.toString());
    }

    public void actualSaveShortLinkStats(ShortLinkStatsRecordDTO statsRecord) {
        // 1. 加锁查询防止短链接分组数据发生改变导致统计数据错误
        String fullShortUrl = statsRecord.getFullShortUrl();
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY, fullShortUrl));
        RLock rLock = readWriteLock.readLock();
        rLock.lock();
        try {
            // 2. 获取gid
            LambdaQueryWrapper<LinkGotoDo> queryWrapper = Wrappers.lambdaQuery(LinkGotoDo.class)
                    .eq(LinkGotoDo::getFullShortUrl, fullShortUrl);
            LinkGotoDo shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);
            String gid = shortLinkGotoDO.getGid();
            // 3. 获取短链接访问时间
            Date currentDate = statsRecord.getCurrentDate();
            int hour = DateUtil.hour(currentDate, true);
            Week week = DateUtil.dayOfWeekEnum(currentDate);
            int weekValue = week.getIso8601Value();
            // 4. 存储基础数据
            LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                    .pv(1)
                    .uv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .uip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .hour(hour)
                    .weekday(weekValue)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkAccessStatsMapper.shortLinkStats(linkAccessStatsDO);
            // 5. 存储ip信息
            Map<String, Object> localeParamMap = new HashMap<>();
            localeParamMap.put("key", statsLocaleAmapKey);
            localeParamMap.put("ip", statsRecord.getRemoteAddr());
            String localeResultStr = HttpUtil.get(AMAP_REMOTE_URL, localeParamMap);
            JSONObject localeResultObj = JSON.parseObject(localeResultStr);
            String infoCode = localeResultObj.getString("infocode");
            String actualProvince = "未知";
            String actualCity = "未知";
            if (StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode, "10000")) {
                String province = localeResultObj.getString("province");
                boolean unknownFlag = StrUtil.equals(province, "[]");
                LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder()
                        .province(actualProvince = unknownFlag ? actualProvince : province)
                        .city(actualCity = unknownFlag ? actualCity : localeResultObj.getString("city"))
                        .adcode(unknownFlag ? "未知" : localeResultObj.getString("adcode"))
                        .cnt(1)
                        .fullShortUrl(fullShortUrl)
                        .country("中国")
                        .date(currentDate)
                        .build();
                linkLocaleStatsMapper.shortLinkLocaleState(linkLocaleStatsDO);
            }
            // 6. 存储操作系统信息
            LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                    .os(statsRecord.getOs())
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkOsStatsMapper.shortLinkOsState(linkOsStatsDO);
            LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                    .browser(statsRecord.getBrowser())
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            // 7. 存储浏览器信息
            linkBrowserStatsMapper.shortLinkBrowserState(linkBrowserStatsDO);
            LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                    .device(statsRecord.getDevice())
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);
            // 8. 存储网络信息
            LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                    .network(statsRecord.getNetwork())
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);
            // 9. 存储访问日志信息
            LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                    .user(statsRecord.getUv())
                    .ip(statsRecord.getRemoteAddr())
                    .browser(statsRecord.getBrowser())
                    .os(statsRecord.getOs())
                    .network(statsRecord.getNetwork())
                    .device(statsRecord.getDevice())
                    .locale(StrUtil.join("-", "中国", actualProvince, actualCity))
                    .fullShortUrl(fullShortUrl)
                    .build();
            linkAccessLogsMapper.insert(linkAccessLogsDO);
            // 10. 更新短链接访问统计信息
            shortLinkMapper.incrementStats(gid, fullShortUrl, 1, statsRecord.getUvFirstFlag() ? 1 : 0, statsRecord.getUipFirstFlag() ? 1 : 0);
            LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                    .todayPv(1)
                    .todayUv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .todayUip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .fullShortUrl(fullShortUrl)
                    .date(currentDate)
                    .build();
            linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);
        } finally {
            // 11. 解锁
            rLock.unlock();
        }
    }

}
